package com.bub.pay.mq.consumer.listener;

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;

import java.util.List;
import java.util.concurrent.*;

public class NotifyDownStreamListener implements BatchMessageListener {

    private ExecutorService notifyDownStreamExecutor = Executors.newCachedThreadPool();



    @Override
    public Action consume(List<Message> messages, ConsumeContext consumeContext) {

        /*List<Future<Action>> result = new LinkedList<>();
        for (Message message : messages) {

            Future<Action> future = notifyDownStreamExecutor.submit(new Callable<Action>() {
                @Override
                public Action call() throws Exception {
                    Map<String, String> map = JSON.parseObject(message.getBody(), Map.class);
                    log.info("通知下游 订单号:{} 请求参数:{} 消息id:{}", map.get("orderNo"), map.toString(), message.getMsgID());
                    String data = null;
                    int i = 0;
                    boolean isSuccess = false;

                    OrganizationNotify organizationNotify = organizationNotifyMapper.selectqueyByOrdNum(map.get("orderNo"));

                    if (organizationNotify != null && organizationNotify.getResultNotifyState() == Constant.NOTIFY_STATE_YES_CODE) {
                        return Action.CommitMessage;
                    }

                    do {
                        i++;
                        try {
                            data = HttpUtils.sendPost(map.get("url"), map, "UTF-8", 1000, 3000);
                        } catch (Exception e) {
                            log.error("通知下游http请求失败 订单号:{} url:{} 请求参数:{} 单次消费请求次数:{} 错误:{} 错误类型:{}", map.get("orderNo"), map.get("url"), map, i, e.getMessage(), e);
                            continue;
                        }

                        log.info("### 商户平台返回 订单号:{} 回调结果:{}", map.get("orderNo"), data);
                        if (data != null && (data.replace(" ", "").equals("0") || data.replace(" ", "").equals("AUTO_SUCCESS"))) {
                            log.info("### 开始修改通知状态: 成功！" + map.get("orderNo"));

                            OrganizationNotify notify = new OrganizationNotify();
                            notify.setOrderNum(map.get("orderNo"));
                            notify.setResultNotifyState(Constant.NOTIFY_STATE_YES_CODE);
                            notify.setResultNotifyCompleteDate(DateUtils.getNow());
                            int notifyresult = organizationNotifyMapper.updateNotifyStatus(notify, null, Constant.NOTIFY_STATE_NO_CODE);
                            isSuccess = true;
                        }
                    } while (i <= 3 && !isSuccess);

                    if (!isSuccess) {
                        //更新失败次数
                        int notifyresult = organizationNotifyMapper.updateNotifyFailCount(map.get("orderNo"));
                        return Action.ReconsumeLater;
                    }
                    return Action.CommitMessage;
                }
            });
            result.add(future);
        }

        for (Future<Action> f : result) {
            try {
                if (f.get(4100, TimeUnit.MILLISECONDS) == Action.ReconsumeLater) {
                    return Action.ReconsumeLater;
                }
            } catch (Exception e) {
                log.error("获取通知下游返回结果超时 进行重试 错误:{}", e.getMessage());
                return Action.ReconsumeLater;
            }
        }*/
        return Action.CommitMessage;
    }
}
